package reactor;

import org.junit.Test;
import org.reactivestreams.Subscription;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;
import reactor.core.publisher.SignalType;
import util.Util;

/**
 * @author yding5
 */
public class SubscriberTest {

    @Test
    public void testLambdaSubscriber() {
        Flux.just(1, 2, 3).subscribe(
            // onNext
            i -> Util.print("onNext: " + i),
            // onError
            e -> e.printStackTrace(),
            // onComplete
            () -> Util.print("onComplete"),
            // onSubscriber
            s -> {
                Util.print("onSubscribe");
                s.request(3);
            });
    }

    @Test
    public void testBaseSubscriber() {
        Flux<Integer> flux = Flux.just(1, 2, 3);
        flux.map(i -> i - 1).subscribe(new BaseSubscriber<>() {

            @Override
            protected void hookOnSubscribe(Subscription s) {
                Util.print("onSubscribe");
                s.request(3);
            }

            @Override
            protected void hookOnNext(Integer i) {
                Util.print("onNext: " + i);
            }

            @Override
            protected void hookOnError(Throwable e) {
                e.printStackTrace();
            }

            @Override
            protected void hookOnComplete() {
                Util.print("onComplete");
            }

            @Override
            protected void hookOnCancel() {
                Util.print("cancel");
            }

            @Override
            protected void hookFinally(SignalType type) {
                Util.print("Finally");
            }
        });
    }
}
